package spider;

import annotation.ClassXpath;
import annotation.FieldXpath;
import bean.DBParams;
import bean.DownloadParams;
import bean.FetchParams;
import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.client.exception.MQBrokerException;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
import com.alibaba.rocketmq.common.message.Message;
import com.alibaba.rocketmq.remoting.exception.RemotingException;
import common.Constant;
import common.Util;
import lombok.Data;
import org.apache.commons.lang3.SerializationUtils;
import org.apache.commons.lang3.text.WordUtils;
import org.jsoup.Jsoup;
import org.jsoup.select.Elements;
import org.openqa.selenium.JavascriptExecutor;
import org.openqa.selenium.WebDriver;
import org.openqa.selenium.phantomjs.PhantomJSDriver;
import org.openqa.selenium.phantomjs.PhantomJSDriverService;
import org.openqa.selenium.remote.CapabilityType;
import org.openqa.selenium.remote.DesiredCapabilities;
import org.openqa.selenium.support.ui.WebDriverWait;
import org.yaml.snakeyaml.Yaml;
import us.codecraft.xsoup.Xsoup;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.Serializable;
import java.lang.annotation.Annotation;
import java.lang.reflect.Field;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.*;
import java.util.function.Function;
import java.sql.*;


/**
 * Created by chen_ on 2017/5/2.
 */
@Data
public class Spider_MQ {
    //tasks cursor && queues cursor
    private int index = 0;
    private static final int PAGE_DEPTH = 10;
    private DefaultMQProducer[] producers = new DefaultMQProducer[PAGE_DEPTH];
    private DefaultMQPushConsumer[] consumers = new DefaultMQPushConsumer[PAGE_DEPTH];

    private Thread[] tasks = new Thread[PAGE_DEPTH];
    private Future[] futures = new Future[PAGE_DEPTH];
    public final String NAME_SRV_ADDR = "localhost:9876";

    private final int PROCESSOR_NUM = 8;
    private ExecutorService taskExecutor = Executors.newFixedThreadPool(PROCESSOR_NUM);

    private Statement stmt;

//    //初始化最开始的url队列
//    public void setQueues0(DefaultMQProducer queues0) {
//        producers[index] = queues0;
//    }

    public void setTask0(Runnable runnable) {
        tasks[0] = new Thread(runnable);
    }

    public static WebDriver initWebDriver() {
        DesiredCapabilities capabilities = DesiredCapabilities.phantomjs();
        capabilities.setCapability("phantomjs.binary.path", Constant.PHANTOMJS_PATH);
        List cliArgsCap = new ArrayList<String>();
        cliArgsCap.add("--web-security=false");
        cliArgsCap.add("--ssl-protocol=any");
        cliArgsCap.add("--ignore-ssl-errors=true");
        cliArgsCap.add("--webdriver-loglevel=INFO");
        cliArgsCap.add("--load-images=false");
        capabilities.setCapability(CapabilityType.SUPPORTS_FINDING_BY_CSS, true);
        capabilities.setCapability(CapabilityType.TAKES_SCREENSHOT, true);
        capabilities.setCapability(PhantomJSDriverService.PHANTOMJS_CLI_ARGS, cliArgsCap);
        return new PhantomJSDriver(capabilities);
    }

//    public static WebDriver initWebDriver() {
//        File file = new File("F:/bin/chromedriver.exe");
//        System.setProperty("webdriver.chrome.driver", file.getAbsolutePath());
//        ChromeOptions chromeOptions = new ChromeOptions();
//        File file2 = new File("C:\\Users\\chen_\\AppData\\Local\\Google\\Chrome\\User Data");
////        chromeOptions.addArguments("user-data-dir="+file2.getAbsolutePath());
//        return  new ChromeDriver(chromeOptions);
//    }

    WebDriver driver = null;

    public boolean isLastThreadLive(int k) {
        if (k > 0) {
            return !futures[k - 1].isDone();
        }
        return false;
    }

    private Spider_MQ run(Function function) {
        consumers[index] = new DefaultMQPushConsumer(String.valueOf(index));
        consumers[index].setNamesrvAddr(NAME_SRV_ADDR);
        try {
            consumers[index].subscribe("Topic" + index, "*");
        } catch (MQClientException e) {
            e.printStackTrace();
        }
        producers[index + 1] = new DefaultMQProducer(String.valueOf(index + 1));
        producers[index + 1].setNamesrvAddr(NAME_SRV_ADDR);
        try {
            producers[index + 1].start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
        int k = index;
        tasks[index + 1] = new Thread(
                () -> {
                    consumers[k].registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
                        try {
                            Object obj = null;
                            try {
                                obj = SerializationUtils.deserialize(msgs.get(0).getBody());
                            } catch (Exception e) {
                                obj = new String(msgs.get(0).getBody());
                            }
                            if ((boolean) function.apply(obj)) {
                                Message msg = new Message("Topic" + (k + 1) /* Topic */,
                                        "TagA" /* Tag */,
                                        SerializationUtils.serialize((Serializable) obj) /* Message body */
                                );
                                producers[k + 1].send(msg);
                            }
                        } catch (InterruptedException e) {
                            System.out.println(e.getMessage());
                        } catch (RemotingException e) {
                            e.printStackTrace();
                        } catch (MQClientException e) {
                            e.printStackTrace();
                        } catch (MQBrokerException e) {
                            e.printStackTrace();
                        }
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    });
                    try {
                        consumers[k].start();
                    } catch (MQClientException e) {
                        e.printStackTrace();
                    }
                }
        );
        index++;
        return this;
    }

    int j = 0;

    public Spider_MQ print() {
        return run(
                (obj) -> {
                    System.out.println(obj + "\t" + j++);
                    return true;
                }
        );
    }

    public Spider_MQ filter(Function func) {
        return run(
                (obj) -> {
                    return (boolean) func.apply(obj);
                }
        );
    }

    public Spider_MQ download(Function func) {
        return run((obj) -> {
            DownloadParams downloadParams = null;
            downloadParams = (DownloadParams) func.apply(obj);
            Util.download(downloadParams.getUrl(), downloadParams.getFilename());
            return true;
        });
    }

    public void end() {
        for (int i = 0; i < index + 1; i++) {
            futures[i] = taskExecutor.submit(tasks[index - i]);
        }
        System.out.println("启动爬虫……");
        taskExecutor.shutdown();
        try {
            taskExecutor.awaitTermination(Long.MAX_VALUE, TimeUnit.MINUTES);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
//        close();
    }

    private void close() {
        try {
            if (stmt != null) {
                conn.close();
            }
        } catch (SQLException se) {
        }// do nothing
        try {
            if (conn != null) {
                conn.close();
            }
        } catch (SQLException se) {
            se.printStackTrace();
        }
    }
    private static boolean flag = true;
    public Spider_MQ save(Class clazz) {
        synchronized (this) {
            if (flag) {
                initDB();
                flag = false;
            }
        }

        return run((obj) -> {
            String tableName = clazz.getSimpleName();
            List<String> insertNames = new ArrayList<>();
            List<String> insertValues = new ArrayList<>();
            for (Field field : clazz.getDeclaredFields()) {
                insertNames.add(field.getName().substring(field.getName().lastIndexOf('.') + 1));
                try {
                    field.setAccessible(true);
                    Object o = field.get(obj);
                    if (!field.getType().isAssignableFrom(Integer.class) && o != null) {
                        insertValues.add("'" + o + "'");
                    } else {
                        insertValues.add((String) o);
                    }
                } catch (IllegalAccessException e) {
                    e.printStackTrace();
                }
            }
            String sql = "insert ignore into " + tableName + "(" + String.join(",", insertNames) + ")" + " values " + "(" + String.join(",", insertValues) + ");";
            try {
//                System.out.println(sql);
                stmt.executeUpdate(sql);
            } catch (SQLException e) {
                e.printStackTrace();
            }
            return true;
        });
    }

    public Spider_MQ chain(Function func) {
        consumers[index] = new DefaultMQPushConsumer(String.valueOf(index));
        consumers[index].setNamesrvAddr(NAME_SRV_ADDR);

        try {
            consumers[index].subscribe("Topic" + index, "*");
        } catch (MQClientException e) {
            e.printStackTrace();
        }
        producers[index + 1] = new DefaultMQProducer(String.valueOf(index + 1));
        producers[index + 1].setNamesrvAddr(NAME_SRV_ADDR);
        try {
            producers[index + 1].start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }

        int k = index;
        tasks[index + 1] = new Thread(
                () -> chain_process(func, k)
        );
        index++;
        return this;
    }

    private void chain_process(Function func, int k) {
        ExecutorService executorService = Executors.newCachedThreadPool();
        consumers[k].registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            Object obj = null;
            try {
                obj = SerializationUtils.deserialize(msgs.get(0).getBody());
            } catch (Exception e) {
                obj = new String(msgs.get(0).getBody());
            }
            FetchParams fetchParams = null;
            fetchParams = (FetchParams) func.apply(obj);
            FetchParams finalFetchParams = fetchParams;
            CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> fetchPageSource(finalFetchParams), executorService);

            future.thenAccept(pageSource -> {
                //使用clazz
                if (finalFetchParams.getClazz() != null) {
                    enqueueWithClazz(k, finalFetchParams, "" + pageSource);
                }
                //不使用clazz
                if (finalFetchParams.getXpath() != null) {
                    List<String> newUrls = Xsoup.compile(finalFetchParams.getXpath()).evaluate(Jsoup.parse("" + pageSource)).list();

                    try {
                        for (int i = 0; i < newUrls.size(); i++) {
                            Message msg = new Message("Topic" + (k + 1) /* Topic */,
                                    "TagA" /* Tag */,
                                    newUrls.get(i).getBytes() /* Message body */
                            );
                            producers[k + 1].send(msg);
                        }
                    } catch (InterruptedException e) {
                        System.out.println(e.getMessage());
                    } catch (RemotingException e) {
                        e.printStackTrace();
                    } catch (MQClientException e) {
                        e.printStackTrace();
                    } catch (MQBrokerException e) {
                        e.printStackTrace();
                    }
                }
            });
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        try {
            consumers[k].start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
//        executorService.shutdown();
//        try {
//            executorService.awaitTermination(Long.MAX_VALUE, TimeUnit.MINUTES);
//        } catch (InterruptedException e) {
//            e.printStackTrace();
//        }
    }

    private void enqueueWithClazz(int k, FetchParams fetchParams, String pageSource) {
        Class clazz = fetchParams.getClazz();
        ClassXpath classXpath = (ClassXpath) clazz.getAnnotation(ClassXpath.class);
        if (classXpath != null) {
            Elements elements = Xsoup.compile(classXpath.value()).evaluate(Jsoup.parse(pageSource)).getElements();
            elements.stream().forEach(
                    element -> {
                        fieldXpathProcess(k + 1, element.outerHtml(), clazz);
                    });
        } else {
            fieldXpathProcess(k + 1, pageSource, clazz);
        }
    }

    private void fieldXpathProcess(int k1, String pageSource, Class clazz) {
        Object obj = null;
        try {
            obj = clazz.newInstance();
        } catch (Exception e) {
            System.out.println(e.getMessage());
        }

        for (Field field : clazz.getDeclaredFields()) {
            Annotation[] annotations = field.getDeclaredAnnotations();
            if (annotations.length != 0) {
                Annotation annotation = annotations[0];
                if (annotation instanceof FieldXpath) {
                    Method setMethod = null;
                    try {
                        setMethod = clazz.getDeclaredMethod("set" + WordUtils.capitalize(field.getName()), String.class);//目前只支持String
                        List<String> lists = Xsoup.compile(((FieldXpath) annotation).value()).evaluate(Jsoup.parse(pageSource)).list();
                        setMethod.invoke(obj, "".join(",", lists));
                    } catch (Exception e) {
                        System.out.println(e.getMessage());
                    }
                }
            }
        }
        try {
            Message msg = new Message("Topic" + k1 /* Topic */,
                    "TagA" /* Tag */,
                    SerializationUtils.serialize((Serializable) obj)
            );
            producers[k1].send(msg);
        } catch (InterruptedException e) {
            System.out.println(e.getMessage());
        } catch (RemotingException e) {
            e.printStackTrace();
        } catch (MQClientException e) {
            e.printStackTrace();
        } catch (MQBrokerException e) {
            e.printStackTrace();
        }

    }

    Random r = new Random();

    public String fetchPageSource(FetchParams fetchParams) {
//        double d2 = r.nextDouble() * 0.5 + 0.5;
//        try {
//            Thread.sleep((long) (d2 * 1000));
//        } catch (InterruptedException e) {
//            e.printStackTrace();
//        }
        String content = "";
        switch (fetchParams.getFlag()) {
            case simple: {
                try {
                    content = Util.req(fetchParams.getUrl());
                } catch (Exception e) {
                    //nothing to do
                }
            }
            break;
            case dynamic: {
                synchronized (this) {
                    try {
                        if (driver == null) {
                            driver = initWebDriver();
                        }
                        driver.get(fetchParams.getUrl());
                        driver.manage().timeouts().implicitlyWait(1, TimeUnit.SECONDS);
                        if (fetchParams.getScript() != null) {
                            try {
                                ((JavascriptExecutor) driver).executeScript(fetchParams.getScript());
                                if (fetchParams.getScriptLocator() != null) {
                                    new WebDriverWait(driver, fetchParams.getTimeout()).until(fetchParams.getScriptLocator());
                                }
                            } catch (Exception e) {
                                System.out.println(e.getMessage());
                            }
                        }
                        content = driver.getPageSource();
                    } catch (Exception e) {
                        System.out.println(e.getMessage());
                    }
                }
            }
        }
        return content;
    }

    Connection conn = null;

    public void initDB() {
        Yaml yaml = new Yaml();
        DBParams me = null;
        try {
            me = yaml.loadAs(new FileInputStream(new File("config/db.yaml")), DBParams.class);
        } catch (FileNotFoundException e) {
            e.printStackTrace();
        }
        System.out.println(me);

        try {
            //STEP 2: Register JDBC driver
            Class.forName("com.mysql.cj.jdbc.Driver");

            //STEP 3: Open a connection
            System.out.println("Connecting to a selected database...");
            conn = DriverManager.getConnection(me.getUrl(), me.getUsername(), me.getPassword());
            conn.setAutoCommit(true);
            System.out.println("Connected database successfully...");

            //STEP 4: Execute a query
            System.out.println("Inserting records into the table...");
            stmt = conn.createStatement();
            stmt.executeUpdate(me.getCreateTableSql());
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
